"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : security-audit
Case Name   : 利用打印函数与索引配合绕过select审计测试
Create At   : 2023/06/08
Owner       : opentestcase033
Description :
    1.设置审计参数
    2.创建测试用户和表，赋权
    3.开启表的行访问控制开关，创建表行访问控制策略
    4.测试用户登录创建打印函数、创建索引嵌入打印函数
    5.查询审计记录
Expect      :
    1.设置成功
    2.成功
    3.成功
    4.成功
    5.查询到创建FUNCTION、INDEX的信息
History     :
"""

import os
import unittest

from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant


class Audit(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.comm = Common()
        self.cons = Constant()
        self.user = 'u_audit_0001'
        self.table = 't_audit_0001'
        self.para_value = self.comm.show_param('audit_system_object')

    def test_security(self):
        text = '-----step1:设置审计参数;expect:成功-----'
        self.log.info(text)
        res = self.pri_sh.execute_gsguc('reload', self.cons.GSGUC_SUCCESS_MSG,
                                        'audit_system_object=67121343')
        self.assertTrue(res, f"执行失败{text}")
        res = self.pri_sh.execute_gsguc('reload', self.cons.GSGUC_SUCCESS_MSG,
                                        'audit_dml_state_select=1')
        self.assertTrue(res, f"执行失败{text}")
        res = self.pri_sh.execute_gsguc('reload', self.cons.GSGUC_SUCCESS_MSG,
                                        'audit_dml_state=1')
        self.assertTrue(res, f"执行失败{text}")

        text = '-----step2:创建用户和表，赋权;expect:成功-----'
        self.log.info(text)
        sql = f'''drop user if exists {self.user} cascade;
        create user {self.user} with password 'test@123';
        create table {self.table}(
        id int, role varchar(100), data varchar(100));
        insert into {self.table} values
        (1, 'alice', 'alice data'), (2, '{self.user}', 'bob data');
        grant select on {self.table} to {self.user};
        grant create any index to {self.user};'''
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertIn(self.cons.CREATE_ROLE_SUCCESS_MSG, msg, f"执行失败:{text}")
        self.assertIn(self.cons.TABLE_CREATE_SUCCESS, msg, f"执行失败:{text}")
        self.assertIn(self.cons.INSERT_SUCCESS_MSG, msg, f"执行失败:{text}")
        self.assertEqual(msg.count(self.cons.GRANT_SUCCESS_MSG), 2,
                         f"执行失败:{text}")

        text = '-----step3:开启表的行访问控制开关，创建表行访问控制策略;expect:成功-----'
        self.log.info(text)
        sql = f"alter table {self.table} enable row level security;" \
              f"create row level security policy all_data_rls " \
              f"on {self.table} using(role = current_user);"
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertIn(self.cons.ALTER_TABLE_MSG, msg, f"执行失败:{text}")
        self.assertIn(self.cons.CREATE_ROW_LEVEL_SECURITY_POLICY_SUCCESS_MSG,
                      msg, f"执行失败:{text}")

        text = '-----step4:创建打印函数、创建索引嵌入打印函数;expect:成功-----'
        self.log.info(text)
        sql = f'''create function raise_demo(id integer,name varchar(100)) 
        returns integer language plpgsql IMMUTABLE as \$\$
        begin
        raise notice 'id :% , name : %',id,name;
        return 1;
        end;
        \$\$;
        create index i_audit_0001 ON {self.table}(raise_demo(1,data));'''
        msg = self.pri_sh.execut_db_sql(
            sql, sql_type=f'-U {self.user} -Wtest@123')
        self.log.info(msg)
        self.assertIn(self.cons.CREATE_FUNCTION_SUCCESS_MSG, msg,
                      f"执行失败:{text}")
        self.assertIn(self.cons.CREATE_INDEX_SUCCESS, msg, f"执行失败:{text}")

        text = '-----step5:查询审计记录;expect:成功-----'
        self.log.info(text)
        sql = f"select pg_sleep(2);" \
              f"select username,type,object_name,detail_info from " \
              f"pg_query_audit(sysdate - interval '20 seconds', sysdate) " \
              f"where username = '{self.user}' and " \
              f"type in ('ddl_function', 'ddl_index');"
        msg = self.pri_sh.execut_db_sql(sql)
        self.log.info(msg)
        self.assertIn(f'create function raise_demo', msg, f"执行失败:{text}")
        self.assertIn('create index i_audit_0001', msg, f"执行失败:{text}")

    def tearDown(self):
        text = '-----step6:清理环境;expect:成功-----'
        self.log.info(text)
        msg = self.pri_sh.execut_db_sql(
            f'drop user if exists {self.user} cascade;'
            f'drop table if exists {self.table} cascade;')
        self.log.info(msg)
        res1 = self.pri_sh.execute_gsguc('reload', self.cons.GSGUC_SUCCESS_MSG,
                                    f'audit_system_object={self.para_value}')
        res2 = self.pri_sh.execute_gsguc('reload', self.cons.GSGUC_SUCCESS_MSG,
                                         'audit_dml_state_select=0')
        self.log.info(res2)
        res3 = self.pri_sh.execute_gsguc('reload', self.cons.GSGUC_SUCCESS_MSG,
                                         'audit_dml_state=0')
        self.assertTrue(res1 and res2 and res3, f"执行失败:{text}")
        self.assertIn(self.cons.TABLE_DROP_SUCCESS, msg, f"执行失败:{text}")
        self.assertIn(self.cons.DROP_ROLE_SUCCESS_MSG, msg, f"执行失败:{text}")
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
